package com.vehicle.mq.queue;

import com.vehicle.common.redis.manager.CacheManager;
import com.vehicle.common.util.Jsons;
import com.vehicle.mq.message.BaseMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

/**
 * Created by yantingbin on 2017/12/1.
 */
public class WaitQueue extends BaseMessageQueue {

    private static final Logger LOGGER = LoggerFactory.getLogger(WaitQueue.class);

    private static final String WAIT_QUEUE_NAME = "disruptor:product";

    private WaitQueue(){

    }

    @Override
    public void push(BaseMessage baseMessage,EventQueue eq) {
        CacheManager cacheManager=eq.getCacheManager();
        cacheManager.lpush(WAIT_QUEUE_NAME, Jsons.toJson(baseMessage));
    }

    @Override
    public BaseMessage pop(EventQueue eq) {
        CacheManager cacheManager=eq.getCacheManager();
        int awaitInmillis= eq.getAwaitInmillis();
        while(true){
            String json= cacheManager.rpop(WAIT_QUEUE_NAME,String.class);
            if(Objects.isNull(json)){
                try {
                    Thread.sleep(awaitInmillis);
                } catch (InterruptedException e) {
                    LOGGER.info("the current thread sleep abnormalities");
                }
            }else{
                  return convert(json);
            }
        }
    }

    public static WaitQueue getInstance() {
        return Single.I;
    }

    private static class Single {
        private static final WaitQueue I = new WaitQueue();
    }
}
